package com.weipt.rabbitmq;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;


/**5
 * 主题模式
 * topic模式，支持多对多绑定队列
 * 后续根据路由key模糊命中后，往相应的队列发消息
 * topic模式在路由模式基础上，增加了模糊匹配
 * 绑定的时候，路由key，可以使用  #和*进行模糊  #：0个，1个，或者多个   *：1个，或者多个（即：至少要有一个）
 * 然后发送消息时，路由key就送实际的
 */
public class HelloTopic {

    public static final String QUEUE_NAME1 = "hello-topic-queue1";

    public static final String EXCHANGE_NAME = "hello-topic-exchange";

    public static final String ROUTING_KEY0 = "topic0*";
    public static final String ROUTING_KEY1 = "topic1.*";

    public static final String ROUTING_KEY2 = "topic2.#";

    public static final String EXCHANGE_TYPE = "topic";

    public static void main(String[] args) throws IOException, TimeoutException {
       // publish("topic1", "test topic");  //没有发出去，因为topic1.* 代表后面至少要有一个
        //publish("topic12", "test topic");  //没有发出去,因为没有.
       // publish("topic01", "test topic");  //没有发出去,*不能和数字直接连接，需要用小数点，不然没有模糊匹配的概念
       // publish("topic0*", "test topic");  //可以收到，所以没有小数点的时候，*和#就是字符
        //publish("topic1.2", "test topic");  //能收到
        publish("topic2", "test topic");  //能收到,因为#是可有可无
        consume();
    }

    public static void publish(String rk, String msg) throws IOException, TimeoutException {
        Connection mqConn = ConnectionUtils.getMqConn();
        Channel channel = mqConn.createChannel(); //创建一个新的管道
        channel.exchangeDeclare(EXCHANGE_NAME, EXCHANGE_TYPE);    //创建一个新的交换机exchange，一般项目中exchange都会提前规划好，此处仅演示
        //一个routingkey可以绑定多个队列
        channel.queueDeclare(QUEUE_NAME1, false, false, false, null);
        channel.queueBind(QUEUE_NAME1, EXCHANGE_NAME, ROUTING_KEY0);
        channel.queueBind(QUEUE_NAME1, EXCHANGE_NAME, ROUTING_KEY1);
        channel.queueBind(QUEUE_NAME1, EXCHANGE_NAME, ROUTING_KEY2);

        //因为rabbitmq是长连接，相当于一个连接中可以创建多个管道就行通讯，类比http长连接
        //所有的消息不能直接由channel发给queue，必须要经过exchange，如果不指定，则走默认的exchage，默认的是direct模式。非direct模式无法走默认
        //发key1,123队列都收到，  发key2,3收到
        channel.basicPublish(EXCHANGE_NAME, rk,  MessageProperties.PERSISTENT_TEXT_PLAIN, msg.getBytes());
        //关闭连接
        channel.close();
        mqConn.close();
    }

    public static void consume() throws IOException {
        Connection mqConn = ConnectionUtils.getMqConn();
        Channel channel = mqConn.createChannel();
        channel.basicConsume(QUEUE_NAME1, true, new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("handleDelivery:"+consumerTag);
                System.out.println("获取queue中的消息。消息内容："+new String(body));
            }
        });
        //消费端连接与同道不用关闭，一直处于监听状态
        //channel.close();
        //mqConn.close();
    }
}
